package system;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.rmi.RMISecurityManager;
import java.rmi.RemoteException;
import java.rmi.registry.LocateRegistry;
import java.rmi.registry.Registry;
import java.rmi.server.UnicastRemoteObject;
import java.util.HashMap;
import java.util.Map;
import java.util.Vector;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import api.Client2Master;
import api.TaskBase;
/**
 * The class for the Master process. It implements the Client2Master interface
 * so that the client can interact with the Master. It also implements the 
 * Worker2Master so that Worker processes can communicate with the Master
 * 
 * @author Mayuri Karra, Bing Wei
 *
 */
public class Master extends UnicastRemoteObject implements Client2Master, Worker2Master{
	private static final long serialVersionUID = 1L;

	protected Master() throws RemoteException {
		super();
	}

	static int TASKID = 0;
	
	Vector<WorkerOnMaster> workers = new Vector<WorkerOnMaster>();
   
	// Input queue for the client to put its task
	BlockingQueue<TaskBase> clientTasks = new LinkedBlockingQueue<TaskBase>();
	
	// result queue for the client to pick up its result
	BlockingQueue<Long> finalResult = new LinkedBlockingQueue<Long>();
	
	// map which tells the number of output files for a given task
	static Map<Long, Integer> numOutputFiles = new HashMap<Long, Integer>();
	
	// TODO not arrayList, but map: workerId-->queue
	BlockingQueue<MasterTaskBase> workerTaskQueue = new LinkedBlockingQueue<MasterTaskBase>();
	BlockingQueue<Long> workerResultQueue = new LinkedBlockingQueue<Long>();
	
	/**
	 * Intermediate class used by the Master process for lookup
	 * 
	 * @author Mayuri Karra, Bing Wei
	 *
	 */
	class LookUpResult{
		LookUpResult(String result){
			this.result = result;
		}
		public String result;
	}
	
	//output queue for lookup results
	BlockingQueue<LookUpResult> lookupQueue = new LinkedBlockingQueue<LookUpResult>();
	
	static int getTASKId(){
		return TASKID++;
	}
	
	/**
	 * getter function which tells the number of alive workers
	 * @return number of alive workers
	 */
	private synchronized int getNumWorkers(){
		int count = 0;
		for(WorkerOnMaster worker: workers){
			if(worker.isAlive())
				count ++;
		}
		return count;
	}
	
	private void addWorker(Worker worker, String hostName, int processId){
		workers.add(new WorkerOnMaster(worker, hostName, processId));
		
	}
	
	private MasterTaskBase getSubTask(){
		try {
			return workerTaskQueue.take();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return null;
	}
	
	private void putSubTask(MasterTaskBase subTask){
		try {
			workerTaskQueue.put(subTask);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	private void putResult(Long i){
		try {
			workerResultQueue.put(i);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	private void putFinalResult(long time){
		finalResult.add(time);
	}
	
	private Long getResult(){
		try {
			return workerResultQueue.take();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return null;
	}
	
	/**
	 * Method to enqueue a map reduce task with 
	 * the MapReduce master
	 * @param task
	 * @throws RemoteException
	 */
	@Override
	public void addTask(TaskBase task) throws RemoteException{
		task.setTaskId(getTASKId());
		clientTasks.add(task);
	}
	
	/**
	 * Remote computers use this to register with the Worker
	 * 
	 * @param computer	
	 *            Reference to a computer that registered with the Space
	 * @throws RemoteException
	 */

	public void register(Worker worker, String hostName, int processId) throws RemoteException {
		
		/*
		 *  instantiate a WorkerProxy, which is a separate thread
		 *  this thread's run method loops forever
		 *  removing tasks from a queue
		 *  invoking the associated Worker's execute method with the task as its argument
		 *  and putting the returned Result object in a data structure for retrieval by the client.
		 */
		
		System.out.println("Received register request "+ hostName);	
		addWorker(worker, hostName, processId);
		/*
		BlockingQueue<MasterTaskBase> taskQueue = new LinkedBlockingQueue<MasterTaskBase>();
		workerTaskQueue.put(taskQueue);
		
		BlockingQueue<MasterTaskBase> resultQueue = new LinkedBlockingQueue<MasterTaskBase>();
		workerResultQueue.put(resultQueue);
		
		Thread workerThread = new Thread(new WorkerProxy(worker, taskQueue, resultQueue));
		*/
		Thread workerThread = new Thread(new WorkerProxy(worker));
		workerThread.start();
		
	} 
	
	public class WorkerProxy implements Runnable {
		Worker worker;
		
		public WorkerProxy(Worker worker) {
			super();
			this.worker = worker;
		}


		@Override
		public void run() {
			while(true){
				MasterTaskBase masterTask = getSubTask();
				long workerTime = 0;
				try {
					// TODO overlap computation with communication
					if(masterTask.getType()!=MasterTaskBase.LOOKUP){
						workerTime = worker.execute(masterTask);
						putResult(workerTime);
					}else{
						String resultStr = worker.lookup(masterTask);
						System.out.println("Returned result " + resultStr);
						putLookupResult(resultStr);
					}					
				} catch (RemoteException e) {
					System.out.println("Detected that a worker died. Put the task back into the queue");
					putSubTask(masterTask);
					break;
				}
			} 
		}
	}
	/*
	private void divideTask(TaskBase task){
		
	}
	*/

	/**
	 * Helper function for dividing input task into sub tasks
	 */
	long divideTask(TaskBase clientTask, int numWorkers){
		
		long timeTaken = 0;
		try {
			String inputFileName = clientTask.getInputFileName();
			
			
			File file = new File(inputFileName);
	
		    if (file.exists()){
		        int lineCount = 0;
		        BufferedReader ln = new BufferedReader(new FileReader(file));
		        String line = null;
		        while (( line = ln.readLine()) != null){
		        	lineCount ++;
		        	//System.out.println(line);
				}
		        //System.out.println(lineCount);
		        //int chunkNum = lineCount/numWorkers+1;
		        int chunkSize = (lineCount/numWorkers)+1;
		        
		        for (int i = 0; i < numWorkers; i++){
		           MasterTaskBase subTask = new MasterTaskBase(0 /* Fix it*/, i, 
		        		                        clientTask,  chunkSize*i, 
		        		                        (((chunkSize* (i+1)) -1) > lineCount) ? lineCount: ((chunkSize* (i+1)) -1), 
		        		                        MasterTaskBase.MAPPER, numWorkers);	   
		           putSubTask(subTask); 
		           System.out.printf("Generate subTask for worker %d\n", i);
		        }
		        
				// get results from all the mappers
		        
			    for (int i = 0; i < numWorkers; i++){
				   timeTaken += getResult();
			       System.out.printf("Got mapper result for a subTask from worker %d\n", i);
			    }
		
		    
		   }
	       else{
	    	   System.out.println("File does not exists!");
	       }
		} catch (FileNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		return timeTaken;
	}
	
	/**
	 * Helper function for dividing the given task
	 * into subtasks for reduce phase
	 * 
	 * @param clientTask
	 * @param numWorkers
	 * @return the time taken in reduce phase
	 */

	long reduceTask(TaskBase clientTask, int numWorkers){
		long timeTaken = 0;
		for (int i = 0; i < numWorkers; i++){
		   MasterTaskBase subTask = new MasterTaskBase(0 /* Fix it*/, i, 
				                        clientTask, 0, 0, 
				                        MasterTaskBase.REDUCER, numWorkers);	   
		   putSubTask(subTask); 
		   System.out.printf("Generate reducer subtask for worker %d\n", i);
		   
		   
		}
		for (int i = 0; i < numWorkers; i++){
			   timeTaken += getResult();
		       System.out.printf("Got reducer result for a subTask from worker %d\n", i);
		    }
	   return timeTaken;
	}
	
	/**
	 * The blocking method which returns when a given task is done 
	 * processing
	 * 
	 * @return the taskId assigned to the task by the task server
	 * @throws RemoteException
	 */
	
	public Long finish() throws RemoteException{
		try {
			return finalResult.take();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return null;
	}
		

	/*
	void shuffle(TaskBase clientTask, int numWorkers){
		try {
		
			// get results from all the mappers
		    for (int i = 0; i < numWorkers; i++){
			   getResult();
		       System.out.printf("Got mapper result for a subTask from worker %d\n", i);
		    }

		    BufferedWriter[] out = new BufferedWriter[numWorkers];
			for (int i = 0; i < numWorkers; i++){
				out[i] = new BufferedWriter(new FileWriter(getReducerInputFileName(i, clientTask.getTaskId())));
			}		
		
		    numOutputFiles.put(clientTask.getTaskId(), numWorkers);
		    for (int i = 0; i < numWorkers; i++){
		        BufferedReader ln = new BufferedReader(new FileReader(getMapperOutputFileName(i, clientTask.getTaskId())));
		        String line = null;
		        while (( line = ln.readLine()) != null){
		        	String[] strs = line.split("\t");
		        	if(strs.length == 2){
		        	   try{
		        		   out[strs[0].hashCode()%numWorkers].write(strs[0] + "\t" + strs[1]+"\n");		        	   		        		
		        	   }catch(Exception e){
		        		   System.out.println("The excepton line is " + line);    
		        	   }
		        	 }
				}
		        ln.close();
		     }
        	     	
			for (int i = 0; i < numWorkers; i++){
				out[i].close();
			}		

		    
		} catch (Exception e) {
			e.printStackTrace();
		}     
	}
*/
	
	/**
	 * Lookup method to lookup a key and return the result
	 * embedded in a String
	 * @param taskId of the task corresponding to this lookup
	 * @param key to lookup[ for
	 * @return the String which has the result
	 * @throws RemoteException
	 */

	public String lookup(long taskId, String key) throws RemoteException{
      
		int numOutput = numOutputFiles.get(taskId);
		System.out.println("Looking up for " + key + " taskId " + taskId);
		for(int i=0; i<numOutput; i++){
			MasterTaskBase subTask = new MasterTaskBase(taskId /* Fix it*/, i, 
                    null, 0, 0, 
                    MasterTaskBase.LOOKUP, key, numOutput);	   
			putSubTask(subTask);
		}
		
		String rst = null;
		try {
			
			for(int i=0; i<numOutput; i++){				
				LookUpResult oneRst = lookupQueue.take();
				if(oneRst.result != null)
					rst = oneRst.result;
			}
			
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		return rst;
	}
	
	
	/**
	 * Method to put the result of lookup into queue
	 * @param rst
	 * @throws RemoteException
	 */
	@Override	
	public void putLookupResult(String rst) throws RemoteException{
		try {
			lookupQueue.put(new LookUpResult(rst));
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	public static void main(String[] args) throws RemoteException{
		try {
			if (System.getSecurityManager() == null) {
	            System.setSecurityManager(new RMISecurityManager());
	        }
			
	        Client2Master master = new Master();
	        Registry registry = LocateRegistry.createRegistry(1099);
	        registry.rebind(Client2Master.SERVICE_NAME, master);
	        registry.rebind(Worker2Master.SERVICE_NAME, master);
	        
	        System.out.println("Master: Master Started !!");
	        
	        while(true){
	        	TaskBase task = ((Master)master).clientTasks.take();
	            int numWorkers = ((Master)master).getNumWorkers();  
	            long timeTaken = 0;
	        	timeTaken += ((Master)master).divideTask(task, numWorkers);
	        	
		        numOutputFiles.put(task.getTaskId(), numWorkers);
//	        	((Master)master).shuffle(task, numWorkers);
	        	
	        	timeTaken += ((Master)master).reduceTask(task, numWorkers);	        	
	        	((Master)master).putFinalResult(timeTaken);
	        }
	        
		} catch (RemoteException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
        
	}

	/**
	 * A method that returns the status of all the worker threads 
	 * @return a vector of Strings which indicate the status of each Worker
	 * @throws RemoteException
	 */
	
	public Vector<String> ping() throws RemoteException{
		Vector<String> workerList = new Vector<String>();
		
		for(WorkerOnMaster aWorker: workers){
			try {				
  				aWorker.getWorker().getPing();
			} catch (RemoteException e) {
				System.out.println("Detected that a worker died");
				aWorker.setStatus (false); /* set it as dead */
			}
			workerList.add(aWorker.getHostName() + " : " + aWorker.getProcessId() + " : " + aWorker.getStatus());
		}
		return workerList;
		
	}
	
	/**
	 * Helper class which stores the state of a Worker process
	 * 
	 * @author Mayuri Karra, Bing Wei
	 *
	 */
	class WorkerOnMaster{
	  Worker worker;   /* reference to worker */
	  String hostName; /* host name of the worker */
	  int processId; /* process id on the host machine */
	  boolean status; /* false - dead ; true - alive */
	  
	  public WorkerOnMaster(Worker worker, String hostName, int processId) {
		super();
		this.worker = worker;
		this.hostName = hostName;
		this.processId = processId;
		this.status = true;
	  }

	  /**
	   * getter function for Worker
	   * @return worker
	   */
	  public Worker getWorker(){
		  return this.worker;
	  }

	  /**
	   * getter function for host name
	   * @return hostName
	   */
	  public String getHostName(){
		  return this.hostName;
	  }
	  
	  /**
	   * getter method for processId
	   * @return processId
	   */
	  public int getProcessId(){
		  return this.processId;
	  }

	  /**
	   * setter function for status
	   * @param status
	   */
	  public void setStatus (boolean status){
	    this.status = status;	  
	  }

	  /**
	   * getter function for status
	   * @return status
	   */
	  public String getStatus(){
		    return (this.status)? "Alive":"Dead";	  
	  }
	  
	  /**
	   * method to tell if a node is alive
	   * @return status
	   */
	  public boolean isAlive(){
		  return this.status;
	  }

	}
	
}
